package com.tensai.study.framework.reactornetty;

import lombok.extern.slf4j.Slf4j;
import reactor.core.Disposable;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.netty.Connection;
import reactor.util.concurrent.Queues;

/**
 * @author Tensai
 */
@Slf4j
public class MyServerConnection {

	final Connection conn;

	final Sinks.Many<String> sinks = Sinks.many().multicast().onBackpressureBuffer(Queues.SMALL_BUFFER_SIZE, false);

	public MyServerConnection(Connection conn) {
		this.conn = conn;
	}

	public void handle() {
		// 从StringToIntegerDecoder接收到作为结果接收到的对象
		conn.inbound().receiveObject()
				.log("服务器连接 - Tensai")
				// 将下一个元素的处理延迟1秒
//				.delayElements(Duration.ofSeconds(1))
				.doOnNext(s ->
						log.info("当前收到的元素: " + s))
				// 在收到五个元素后取消（实际上断开了客户端的连接）
				.take(50)
				.flatMap(s ->
						conn.outbound().sendString(
								Mono.just(String.format("字节统计: %s", s.toString()))
						).then()
				)
				.subscribe(conn.disposeSubscriber());
		// 我们必须按顺序使用该核心订阅者，以使连接在take（5）之后断开连接
		conn.outbound()
		    .sendString(sinks.asFlux())
				.then()
		    .subscribe();
	}
}
